package com.chukun.flink.stream.operator.base;

import org.apache.flink.streaming.api.datastream.ConnectedStreams;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.co.CoMapFunction;

import java.util.ArrayList;

/**
 * @author chukun
 * @version 1.0.0
 * @description 将两个不同类型的数据流合并
 * @createTime 2022年05月11日 00:33:00
 */
public class CoMapFunctionOperator {

    public static void main(String[] args) throws Exception {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        DataStream<Long> longStream = env.fromCollection(new ArrayList<Long>() {
            {
                add(10L);
                add(100L);
            }
        });

        DataStream<String> strStream = env.fromCollection(new ArrayList<String>() {
            {
                add("hello world");
                add("hello flink");
                add("streaming");
            }
        });

        // 拼接流
        ConnectedStreams<Long, String> connectStreams = longStream.connect(strStream);

        // 合并成一个数据流
        DataStream<String> mapStream = connectStreams.map(new CoMapFunction<Long, String, String>() {

            @Override
            public String map1(Long aLong) throws Exception {
                return "long -->" + aLong;
            }

            @Override
            public String map2(String s) throws Exception {
                return "string-->" + s;
            }
        });

        mapStream.print("输出合并流");

        env.execute("CoMapFunction operator");
    }
}
